package cn.shadow.concurrent;


import cn.shadow.message.MessageProcessor;
import cn.shadow.utils.ObjectHolder;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import io.netty.buffer.ByteBuf;

public class DisruptorExecutor implements Executor{
    private final int id;
    private final Disruptor<Event> disruptor;
    private final RingBuffer<Event> ringBuffer;
    private final ExecutorService executorService;
    private final Thread thread;

    public DisruptorExecutor(int id, int bufferSize, String name) {
        if (!checkSize(bufferSize)) {
            throw new IllegalArgumentException("bufferSize 必须为2的倍数");
        } else {
            this.id = id;
            this.executorService = Executors.newSingleThreadExecutor(new DefaultThreadFactory(name));
            ObjectHolder<Thread> holder = new ObjectHolder<>(null);
            CountDownLatch countDownLatch = new CountDownLatch(1);
            this.executorService.execute(new Task(this, holder, countDownLatch));
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            this.thread = holder.get();
            if (this.thread == null) {
                throw new RuntimeException("Thread = null");
            } else {
                this.disruptor = new Disruptor<>(new DefaultEventFactory(), bufferSize, this.executorService, ProducerType.MULTI, new YieldingWaitStrategy());
                MyEventHandler[] eventHandlers = new MyEventHandler[]{new MyEventHandler()};
                this.disruptor.handleEventsWith(eventHandlers);
                this.ringBuffer = this.disruptor.start();
            }
        }
    }

    public int getId() {
        return this.id;
    }

    public Thread getThread() {
        return this.thread;
    }


    private static boolean checkSize(int size) {
        return size > 0 && (size & size - 1) == 0;
    }

    public void execute(Runnable task) {
        long next = this.ringBuffer.next();
        try {
            Event event = this.ringBuffer.get(next);
            event.setTask(task);
        } finally {
            this.ringBuffer.publish(next);
        }
    }

    public boolean tryExecute(Runnable task) {
        try {
            long next = this.ringBuffer.tryNext();
            try {
                Event event = this.ringBuffer.get(next);
                event.setTask(task);
            } finally {
                this.ringBuffer.publish(next);
            }

            return true;
        } catch (InsufficientCapacityException var9) {
            return false;
        }
    }

    public void processMessage(ByteBuf buf, int moduleID, int sequenceID, MessageProcessor processor) {
        long next = this.ringBuffer.next();

        try {
            Event event = this.ringBuffer.get(next);
            event.setEvent(buf, moduleID, sequenceID, processor);
        } finally {
            this.ringBuffer.publish(next);
        }

    }

    public boolean tryProcessMessage(ByteBuf buffer, int moduleID,
                                     int sequenceID, MessageProcessor processor) {
        try {
            long next = this.ringBuffer.tryNext();

            try {
                Event event = this.ringBuffer.get(next);
                event.setEvent(buffer, moduleID, sequenceID, processor);
            } finally {
                this.ringBuffer.publish(next);
            }
            return true;
        } catch (InsufficientCapacityException var12) {
            return false;
        }
    }

    public void shutdown() {
        this.disruptor.shutdown();
        this.executorService.shutdown();
        try {
            this.executorService.awaitTermination(1, TimeUnit.DAYS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}
